package org.atlantis.ymer.server;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.atlantis.ymer.Disposable;
import org.atlantis.ymer.WebSessionFactory;
import org.atlantis.ymer.YmerOptions;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.http.HttpChunkAggregator;
import org.jboss.netty.handler.codec.http.HttpRequestDecoder;
import org.jboss.netty.handler.codec.http.HttpResponseEncoder;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConnectorServer implements Disposable {

	private static final Logger log = LoggerFactory.getLogger(ConnectorServer.class);
	private ServerBootstrap bootstrap;
	private Channel serverChannel;
	private ConnectorHandler connectorHandler;
	private int port;

	private final ExecutorService handlerExecutor;

	public ConnectorServer(WebSessionFactory sessionFactory, YmerOptions options) {
		this.connectorHandler = new ConnectorHandler(sessionFactory);
		this.port = options.getConnectorServerPort();
		this.handlerExecutor = Executors.newFixedThreadPool(options.getHandlerThreadSize());
	}

	public void start() {
		bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool()));
		bootstrap.setPipelineFactory(new ConnectorPipelineFactory());
		bootstrap.setOption("child.keepAlive", true);
		bootstrap.setOption("child.tcpNoDelay", true);
		serverChannel = bootstrap.bind(new InetSocketAddress(port));
		log.info("Ymer ConnectorServer started! Listening on {}", port);
	}

	private class ConnectorPipelineFactory implements ChannelPipelineFactory {

		private ExecutionHandler executionHandler = new ExecutionHandler(handlerExecutor);

		@Override
		public ChannelPipeline getPipeline() throws Exception {
			ChannelPipeline pipeline = Channels.pipeline();
			pipeline.addLast("http-decoder", new HttpRequestDecoder());
			pipeline.addLast("http-aggregator", new HttpChunkAggregator(65536));
			pipeline.addLast("http-encoder", new HttpResponseEncoder());
			pipeline.addLast("execution-handler", executionHandler);
			pipeline.addLast("connector-handler", connectorHandler);
			return pipeline;
		}
	}

	@Override
	public void dispose() {
		serverChannel.close();
		bootstrap.releaseExternalResources();
		handlerExecutor.shutdownNow();
	}
}
